#include <stdexcept>
#include <type_traits>
#include "module_frame.hh"
#include "framework.hh"
#include "entity.hh"
#include "entity_search_cache.hh"
#include "flatbuffers_helper.hh"
#include "module_frame_http.hh"
#include "../protocol/protocol_register.hh"
#include "../protocol/protocol_header.hh"
#include "../protocol/locator/register_entity_req_generated.h"
#include "../protocol/locator/register_entity_ack_generated.h"
#include "../protocol/locator/unregister_entity_req_generated.h"
#include "../protocol/locator/unregister_entity_ack_generated.h"
#include "../protocol/locator/entity_relay_msg_req_generated.h"
#include "../protocol/locator/entity_relay_msg_ack_generated.h"
#include "../protocol/locator/entity_search_req_generated.h"
#include "../protocol/locator/entity_search_ack_generated.h"
#include "../protocol/locator/entity_search_ntf_generated.h"
#include "../../corelib/redis/redis.hh"
#include "../../corelib/mysql/mysql.hh"
#include "../../corelib/network/parameter.hh"
#include "../../corelib/time_util.hh"
#include "../../corelib/event/event_queue.hh"
#include "../../corelib/grid/thirdparty/flatbuffers/flatbuffers.h"

Engine * ModuleFrame::getEngine() {
    if (moduleType_ == ModuleType::SERVICE_MODULE) {
        return nullptr;
    }
    return engine_;
}

bool ModuleFrame::acknowledge(PathID pathID, std::uint32_t msgID, flatbuffers::FlatBufferBuilder& builder) {
    ProtocolLocatorHeader::toBuffer(sendBuffer_.get(), sizeof(ProtocolLocatorHeader) + builder.GetSize(),
        msgID, builder.GetSize());
    memcpy(sendBuffer_.get() + sizeof(ProtocolLocatorHeader), reinterpret_cast<char*>(builder.GetBufferPointer()),
        builder.GetSize());
    return engine_->talkPath(pathID, sendBuffer_.get(), sizeof(ProtocolLocatorHeader) + builder.GetSize());
}

bool ModuleFrame::acknowledge(const PathIDVector & pathIDs, std::uint32_t msgID, flatbuffers::FlatBufferBuilder & builder) {
    buildLocatorHeaderInSendBuffer(0, msgID, builder.GetSize(), sizeof(ProtocolLocatorHeader) + builder.GetSize());
    memcpy(sendBuffer_.get() + sizeof(ProtocolLocatorHeader), reinterpret_cast<char*>(builder.GetBufferPointer()),
        builder.GetSize());
    for (auto pathID : pathIDs) {
        if (!engine_->talkPath(pathID, sendBuffer_.get(), sizeof(ProtocolLocatorHeader) + builder.GetSize())) {
            warnlog << "Send entity un/registering event failed" << endlog;
        }
    }
    return true;
}

bool ModuleFrame::searchEntity(std::uint32_t entityType, std::uint32_t maxCount, std::uint32_t searchType) {
    static constexpr std::size_t MAX_BUFFER_SIZE = 256;
    char buffer[MAX_BUFFER_SIZE] = { 0 };
    BUILD_FLATBUFFER(EntitySearchReq, fb, entityType, maxCount, searchType);
    auto bytes = sizeof(ProtocolLocatorHeader) + static_cast<std::uint32_t>(fb.GetSize());
    ProtocolLocatorHeader::toBuffer(buffer, bytes, getMsgRealID(LocatorMsgID::SEARCH_REQ), fb.GetSize());
    memcpy(buffer + sizeof(ProtocolLocatorHeader), reinterpret_cast<char*>(fb.GetBufferPointer()), fb.GetSize());
    return engine_->broadcastLoactor(buffer, bytes);
}

void ModuleFrame::addHttpDeadCb(InternalHttpCallback * cb) {
    httpDeadCbList_.push_back(cb); 
}

void ModuleFrame::disposeAllHttpDeadCb() {
    for (auto cb : httpDeadCbList_) {
        kratos::corelib::dispose(cb);
    }
    httpDeadCbList_.clear();
}

void ModuleFrame::returnSearchResult(PathID pathID, EntityIDVector& entityIDs, std::uint32_t entityType,
    std::size_t count, std::size_t total) {
    static flatbuffers::FlatBufferBuilder builder(kratos::NETWORK_IOLOOP_RECV_BUFFER_MAX_SIZE);
    builder.Clear();
    auto entities = builder.CreateVector<std::uint64_t>(entityIDs);
    auto ack = CreateEntitySearchAck(builder, entities, entityType, count, total);
    builder.Finish(ack);
    FRAMEWORK_CHECK(builder.GetSize() + sizeof(ProtocolLocatorHeader) > kratos::NETWORK_IOLOOP_RECV_BUFFER_MAX_SIZE,
        "Locator error, send message LOCATOR_MSG_ID_ENTITY_SEARCH_ACK failed, packet size:" << builder.GetSize() + sizeof(ProtocolLocatorHeader));
    FRAMEWORK_CHECK(!acknowledge(pathID, static_cast<std::uint32_t>(LocatorMsgID::SEARCH_ACK), builder),
        "Locator error, send message LOCATOR_MSG_ID_ENTITY_SEARCH_ACK failed");
}

void ModuleFrame::doLocatorMessage(PathID pathID, std::uint32_t msgID, const char * data, std::size_t size) {
    switch (static_cast<LocatorMsgID>(msgID)) {
        case LocatorMsgID::RELAY_REQ:
            onRelayReq(pathID, msgID, data, size);
            break;
        case LocatorMsgID::RELAY_ACK:
            onRelayAck(pathID, msgID, data, size);
            break;
        case LocatorMsgID::REGISTER_REQ:
            onRegisterReq(pathID, msgID, data, size);
            break;
        case LocatorMsgID::REGISTER_ACK:
            onRegisterAck(pathID, msgID, data, size);
            break;
        case LocatorMsgID::UNREGISTER_REQ:
            onUnregisterReq(pathID, msgID, data, size);
            break;
        case LocatorMsgID::UNREGISTER_ACK:
            onUnregisterAck(pathID, msgID, data, size);
            break;
        case LocatorMsgID::SEARCH_REQ:
            onSearchReq(pathID, msgID, data, size);
            break;
        case LocatorMsgID::SEARCH_ACK:
            onSearchAck(pathID, msgID, data, size);
            break;
        case LocatorMsgID::SEARCH_NTF:
            onSearchNtf(pathID, msgID, data, size);
            break;
        default:
            errorlog << "Locator message not found, message ID:" << msgID << endlog;
            break;
    }
}

void ModuleFrame::onRelayReq(PathID pathID, std::uint32_t, const char * data, std::size_t size) {
    // receives a relay message to entity
    ProtocolLocatorHeader locatorHeader(data);
    auto LocatorHeaderLength = locatorHeader.locatorMsgLength + sizeof(ProtocolLocatorHeader);
    FRAMEWORK_CHECK(locatorHeader.length > kratos::NETWORK_IOLOOP_RECV_BUFFER_MAX_SIZE,
        "Invalid relay packet size:" << locatorHeader.length);
    FRAMEWORK_CHECK(LocatorHeaderLength > locatorHeader.length,
        "Invalid relay packet size, locator packet size:" << locatorHeader.locatorMsgLength);
    GET_FLATBUFFER_OBJ(EntityRelayMsgReq, req, data + sizeof(locatorHeader));
    auto entity = EntityManagerRef.getEntity(req->targetEntityID());
    if (entity) {
        bool success = true;
        ProtocolHeader header(data + LocatorHeaderLength);
        try {
            auto msg = ProtocolRegisterRef.getMessage(header.msgID);
            if (msg) {
                if (!msg->ParseFromArray(data + LocatorHeaderLength + sizeof(header), header.length - sizeof(header))) {
                    errorlog << "ParseFromArray failed, message name:" << msg->GetDescriptor()->name() << endlog;
                } else {
                    auto handler = ProtocolRegisterRef.getHandler(header.msgID);
                    if (handler) {
                        handler->onEntityMessage(req->sourceEntityID(), req->targetEntityID(), *msg);
                    } else {
                        framework_->onEntityData(req->sourceEntityID(), req->targetEntityID(), header.msgID, *msg);
                    }
                }
            } else {
                success = false;
            }                
        } catch (std::exception& e) {
            errorlog << "Exception thrown by onEntityData, reason:" << e.what() << endlog;
        }
        BUILD_FLATBUFFER(EntityRelayMsgAck, fb, req->targetEntityID(), success);
        if (!acknowledge(pathID, getMsgRealID(LocatorMsgID::RELAY_ACK), fb)) {
            errorlog << "Locator error, send message LOCATOR_MSG_ID_RELAY_ACK failed" << endlog;
        }
    } else {
        auto entityPathID = engine_->getEntityPathID(req->targetEntityID());
        if (INVALID_PATH_ID == entityPathID) {
            //errorlog << "Entity's path ID NOT found, entity ID:" << req->targetEntityID() << logend;
            BUILD_FLATBUFFER(EntityRelayMsgAck, fb, req->targetEntityID(), false);
            if (!acknowledge(pathID, getMsgRealID(LocatorMsgID::RELAY_ACK), fb)) {
                errorlog << "Locator error, send message LOCATOR_MSG_ID_RELAY_ACK failed" << endlog;
            }
        } else {
            if (!engine_->talkPath(entityPathID, data, size)) {
                //errorlog << "Locator error, relay message to target failed, path ID:" << entityPathID << endlog;
            }
        }
    }
}

void ModuleFrame::onRelayAck(PathID, std::uint32_t, const char * data, std::size_t) {
    GET_FLATBUFFER_OBJ(EntityRelayMsgAck, ack, data + sizeof(ProtocolLocatorHeader));
    //FRAMEWORK_CHECK(!ack->success(), "Locator relay failed target entity ID:" << ack->targetEntityID());
}

void ModuleFrame::onRegisterReq(PathID pathID, std::uint32_t, const char * data, std::size_t) {
    // Add entity to engine
    ProtocolLocatorHeader header(data);
    GET_FLATBUFFER_OBJ(RegisterEntityReq, req, data + sizeof(header));
    static flatbuffers::FlatBufferBuilder builder;
    builder.Clear();
    if (!engine_->addEntity(pathID, req->entityID(), req->entityType())) {
        //errorlog << "Register entity failed, entity ID:" << req->entityID() << logend;
        auto ack = CreateRegisterEntityAck(builder, req->entityID(), false);
        builder.Finish(ack);
    } else {
        auto ack = CreateRegisterEntityAck(builder, req->entityID(), true);
        builder.Finish(ack);
        notifyCacheAdd(req->entityType(), req->entityID());
    }
    FRAMEWORK_CHECK(!acknowledge(pathID, getMsgRealID(LocatorMsgID::REGISTER_ACK), builder),
        "Locator error, send message LOCATOR_MSG_ID_REGISTER_ACK failed");
}

void ModuleFrame::onRegisterAck(PathID pathID, std::uint32_t, const char * data, std::size_t) {
    GET_FLATBUFFER_OBJ(RegisterEntityAck, ack, data + sizeof(ProtocolLocatorHeader));
    if (!ack->success()) {
        //errorlog << "Locator register failed for entity ID:" << ack->entityID() << endlog;
    } else {
        // The registering process finished
        framework_->onFinishRegister(pathID, ack->entityID());
    }
}

void ModuleFrame::onUnregisterReq(PathID pathID, std::uint32_t, const char * data, std::size_t) {
    // Remove entity from engine
    ProtocolLocatorHeader header(data);
    GET_FLATBUFFER_OBJ(UnregisterEntityReq, req, data + sizeof(header));
    static flatbuffers::FlatBufferBuilder builder;
    builder.Clear();
    auto entityType = engine_->getEntityType(req->entityID());
    FRAMEWORK_CHECK(!entityType, "Entity type not found, entity ID:" << req->entityID());
    if (!engine_->removeEntity(pathID, req->entityID())) {
        errorlog << "Unregister entity failed, entity ID:" << req->entityID() << logend;
        auto ack = CreateUnregisterEntityAck(builder, req->entityID(), false);
        builder.Finish(ack);
    } else {
        auto ack = CreateUnregisterEntityAck(builder, req->entityID(), true);
        builder.Finish(ack);
        notifyCacheRemove(entityType, req->entityID());
    }
    FRAMEWORK_CHECK(!acknowledge(pathID, getMsgRealID(LocatorMsgID::UNREGISTER_ACK), builder),
        "Locator error, send message LOCATOR_MSG_ID_UNREGISTER_ACK failed");
}

void ModuleFrame::onUnregisterAck(PathID, std::uint32_t, const char * data, std::size_t) {
    GET_FLATBUFFER_OBJ(UnregisterEntityAck, ack, data + sizeof(ProtocolLocatorHeader));
    FRAMEWORK_CHECK(!ack->success(), "Locator unregister failed for entity ID:" << ack->entityID());
}

void ModuleFrame::onSearchReq(PathID pathID, std::uint32_t, const char * data, std::size_t) {
    GET_FLATBUFFER_OBJ(EntitySearchReq, req, data + sizeof(ProtocolLocatorHeader));
    typePathMap_[req->entityType()].push_back(pathID);
    auto& entityList = engine_->getEntities(req->entityType());
    auto maxCount = static_cast<std::size_t>(req->maxCount());
    if (!maxCount) {
        maxCount = static_cast<std::size_t>(-1);
    }
    static constexpr std::size_t MAX_COUNT_EACH_MESSAGE = 128;
    std::size_t count = 0;
    std::size_t totalCount = entityList.size();
    static EntityIDVector entityIDs;
    entityIDs.clear();
    for (auto& entityID : entityList) {
        if (++count > maxCount) {
            break;
        }
        entityIDs.push_back(entityID);
        if (entityIDs.size() >= MAX_COUNT_EACH_MESSAGE) {
            returnSearchResult(pathID, entityIDs, req->entityType(), count, totalCount);
            entityIDs.clear();
        }
    }
    if (!entityIDs.empty()) {
        returnSearchResult(pathID, entityIDs, req->entityType(), count, totalCount);
    } else {
        if (!count) {
            returnSearchResult(pathID, entityIDs, req->entityType(), count, totalCount);
        }
    }
}

void ModuleFrame::onSearchAck(PathID, std::uint32_t, const char * data, std::size_t) {
    GET_FLATBUFFER_OBJ(EntitySearchAck, ack, data + sizeof(ProtocolLocatorHeader));
    static EntityIDVector entityIDs;
    entityIDs.clear();
    for (auto entityID : *ack->entities()) {
        entityIDs.push_back(entityID);
    }
    // update search cache
    EntitySearchCacheManagerRef.update(ack->entityType(), entityIDs, CacheOP::ADD);
}

void ModuleFrame::onSearchNtf(PathID, std::uint32_t, const char * data, std::size_t) {
    GET_FLATBUFFER_OBJ(EntitySearchNtf, ntf, data + sizeof(ProtocolLocatorHeader));
    static EntityIDVector entityIDs;
    entityIDs.clear();
    for (auto entityID : *ntf->entities()) {
        entityIDs.push_back(entityID);
    }
    // update search cache
    EntitySearchCacheManagerRef.update(ntf->entityType(), entityIDs, static_cast<CacheOP>(ntf->type()));
}

void ModuleFrame::notifyCacheAdd(std::uint32_t entityType, std::uint64_t entityID) {
    notifyCacheOP(entityType, entityID, CacheOP::ADD);
}

void ModuleFrame::notifyCacheRemove(std::uint32_t entityType, std::uint64_t entityID) {
    notifyCacheOP(entityType, entityID, CacheOP::REMOVE);
}

void ModuleFrame::notifyCacheOP(std::uint32_t entityType, std::uint64_t entityID, CacheOP cacheOP) {
    static flatbuffers::FlatBufferBuilder builder;
    static EntityIDVector entityIDs;
    builder.Clear();
    entityIDs.clear();
    entityIDs.push_back(entityID);
    auto entities = builder.CreateVector<std::uint64_t>(entityIDs);
    auto ntf = CreateEntitySearchNtf(builder, entityType, entities, static_cast<std::uint32_t>(cacheOP));
    auto& paths = typePathMap_[entityType];
    builder.Finish(ntf);
    FRAMEWORK_CHECK(!acknowledge(paths, getMsgRealID(LocatorMsgID::SEARCH_NTF), builder),
        "Locator notify cache event failed");
}

ProtocolLocatorHeader * ModuleFrame::buildLocatorHeaderInSendBuffer(std::size_t offset, std::uint32_t msgID,
    std::uint32_t locatorMsgLength, std::uint32_t length) {
    auto locatorHeader  = reinterpret_cast<ProtocolLocatorHeader*>(sendBuffer_.get() + offset);
    locatorHeader->msgID = msgID;
    locatorHeader->locatorMsgLength = locatorMsgLength;
    locatorHeader->length = length;
    locatorHeader->hton();
    return locatorHeader;
}

const std::string& ModuleFrame::getRootPath() {
    return engine_->getPluginRootPath(serviceName_);
}
